package com.zdb.demo.java.flink;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class LoginFail {

    public static void main(String[] args) {

        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(4);

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.fromCollection(Lists.newArrayList(
                new LoginEvent(1L, "192.168.0.1", "fail", 1558430842L),
                new LoginEvent(1L, "192.168.0.2", "fail", 1558430843L),
                new LoginEvent(1L, "192.168.0.3", "fail", 1558430844L),
                new LoginEvent(2L, "192.168.10.1", "fail", 1558430845L)

        )).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<LoginEvent>() {
            @Override public long extractAscendingTimestamp(LoginEvent element) {
                return element.eventTime * 1000;
            }
        }).keyBy(v -> v.userId)
                .process(new KeyedProcessFunction<Long, LoginEvent, Warning>() {

                    ListState<LoginEvent> loginEventListState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        loginEventListState = getRuntimeContext().getListState(
                                new ListStateDescriptor<LoginEvent>("login-state", LoginEvent.class));
                    }

                    @Override
                    public void processElement(LoginEvent value, Context ctx,
                            Collector<Warning> out) throws Exception {
                        if("fail".equals(value.eventType)) {
                            loginEventListState.add(value);
                            // 2秒检测
                            ctx.timerService().registerEventTimeTimer(value.eventTime*1000 + 2000);
                        } else {
                            loginEventListState.clear();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Warning> out)
                            throws Exception {
                        Iterator<LoginEvent> iter = loginEventListState.get().iterator();
                        List<LoginEvent> list = new ArrayList<>();
                        while (iter.hasNext()) {
                            list.add(iter.next());
                        }
//                        loginEventListState.clear();
                        if(list.size() > 1) {
                            out.collect(new Warning(list.get(0).userId, list.get(0).eventTime, list.get(list.size()-1).eventTime, "ligin fail in 2 seconds for " + list.size() + " times"));
                        }
                    }
                }).print();

        try {
            env.execute("login fail detect");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


class LoginEvent {

    Long userId;
    String ip;
    String eventType;
    Long eventTime;

    public LoginEvent(Long userId, String ip, String eventType, Long eventTime) {
        this.userId = userId;
        this.ip = ip;
        this.eventType = eventType;
        this.eventTime = eventTime;
    }

    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public String getEventType() {
        return eventType;
    }

    public void setEventType(String eventType) {
        this.eventType = eventType;
    }

    public Long getEventTime() {
        return eventTime;
    }

    public void setEventTime(Long eventTime) {
        this.eventTime = eventTime;
    }
}

class Warning {
    Long userId;
    Long firstFailTime;
    Long lastFailTime;
    String warningMsg;

    public Warning(Long userId, Long firstFailTime, Long lastFailTime, String warningMsg) {
        this.userId = userId;
        this.firstFailTime = firstFailTime;
        this.lastFailTime = lastFailTime;
        this.warningMsg = warningMsg;
    }

    @Override public String toString() {
        return "Warning{" +
                "userId=" + userId +
                ", firstFailTime=" + firstFailTime +
                ", lastFailTime=" + lastFailTime +
                ", warningMsg='" + warningMsg + '\'' +
                '}';
    }
}